package com.jscloud.bigdata.flink.flinksql.overwindows;

import com.jscloud.bigdata.flink.tableapi.groupwindow.TempSensorData;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * # 基于eventTime向前3行开窗
 * 数据内容如下：
 * s-5,1645085900,14,
 * s-5,1645085901,17,
 * s-5,1645085902,22,
 * s-5,1645085903,7,
 * s-5,1645085904,21,
 * s-5,1645085905,23,
 * s-5,1645085906,8,
 * s-5,1645085907,32,
 * s-5,1645085908,15,
 * s-5,1645085909,9,
 */
public class FlinkSQLOverWinEventRowRanger {
        public static void main(String[] args) {
                //1.获取stream的执行环境
                StreamExecutionEnvironment senv= StreamExecutionEnvironment.getExecutionEnvironment();
                senv.setParallelism(1);
                //2.创建表执行环境
                StreamTableEnvironment tEnv = StreamTableEnvironment.create(senv);

                //3.读取数据
                WatermarkStrategy<TempSensorData> watermarkStrategy = WatermarkStrategy
                        .<TempSensorData>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner(new SerializableTimestampAssigner<TempSensorData>() {
                                @Override
                                public long extractTimestamp(TempSensorData t, long l) {
                                        return t.getTp()*1000;
                                }
                        });
                DataStream<TempSensorData> tempSensorData=senv.socketTextStream("bigdata01",8888)
                        .map(event -> {
                                String[] arr = event.split(",");
                                return TempSensorData.builder()
                                        .sensorID(arr[0])
                                        .tp(Long.parseLong(arr[1]))
                                        .temp(Integer.parseInt(arr[2]))
                                        .build();
                        }).assignTimestampsAndWatermarks(watermarkStrategy);

                //4.流转换为动态表
                Table table = tEnv.fromDataStream(tempSensorData,
                        $("sensorID"),$("tp"),$("temp"),$("evTime").rowtime());

                //5.自定义窗口并计算
                Table resultTable = tEnv.sqlQuery("select "+
                        "sensorID,"+
                        "max(temp) OVER w AS max_temp,"+
                        "avg(temp) OVER w AS avg_temp "+
                        "from "+table+" WINDOW w AS (\n" +
                        " PARTITION BY sensorID\n" +
                        " ORDER BY evTime\n" +
                        " ROWS BETWEEN 3 PRECEDING AND CURRENT ROW)\n");
                //6.执行Flink
                resultTable.execute().print();
        }
}